/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.common.io;

import static org.sss.common.Utils.wrap;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.common.CloseableIterator;
import org.sss.common.DataReceiver;
import org.sss.common.KeyValues;
import org.sss.mapreduce.SssException;

public class KeyValuesQueue {
  @SuppressWarnings("unused")
  private static Logger logger = LoggerFactory.getLogger(KeyValuesQueue.class);

  private static int INITIAL_VALUE_LIST_LENGTH = 128;

  private List<KeyValues>       kvs = null;
  private byte[]                key = null;
  private List<byte[]>          values = null;
  private int                   limitNBytes;
  private int                   cachedNbytes = 0;
  private ValueQueue            valueQueue = null;
  DataReceiver<List<KeyValues>> next;

  public KeyValuesQueue(DataReceiver<List<KeyValues>> next, int limitNBytes) {
    this.limitNBytes = limitNBytes;
    this.kvs = new ArrayList<KeyValues>();
    this.next = next;
  }

  public void pushKey(byte[] key) throws SssException, InterruptedException {
    endValue();
    if (kvs == null) {
      kvs = new ArrayList<KeyValues>();
    }
    this.key = key;
    this.values = new ArrayList<byte[]>(INITIAL_VALUE_LIST_LENGTH);
  }

  public void pushValue(byte[] value) throws SssException, InterruptedException {
    values.add(value);
    cachedNbytes += key.length + value.length;
    if (cachedNbytes >= limitNBytes) {
      if (kvs != null) {
        valueQueue = new ValueQueue(values);
        kvs.add(new KeyValues(key, valueQueue.iterator()));
        next.put(kvs);
        kvs = null;
      }
      else {
        valueQueue.add(values);
      }
      cachedNbytes = 0;
      values = new ArrayList<byte[]>(INITIAL_VALUE_LIST_LENGTH);
    }
  }

  void pushEnd() throws SssException, InterruptedException {
    endValue();
    if (kvs != null) {
      next.put(kvs);
    }
  }

  private void endValue() throws SssException, InterruptedException {
    if (values != null && !values.isEmpty()) {
      if (kvs != null) {
        kvs.add(new KeyValues(key, wrap(values.iterator())));
      }
      else {
        valueQueue.add(values);
      }
    }
    if (valueQueue != null) {
      valueQueue.close();
      valueQueue = null;
    }
    values = null;
  }

  private static class ValueQueue {
    private List<byte[]> next = null;
    private boolean done = false;
    private boolean noMoreRead = false;

    public ValueQueue(List<byte[]> firstList) {
      this.next = firstList;
    }

    public synchronized void add(List<byte[]> values) throws InterruptedException {
      while (next != null && !noMoreRead) {
        wait();
      }
      if (noMoreRead) {
        return;
      }
      else {
        next = values;
        notifyAll();
      }
    }

    public synchronized void close() throws InterruptedException {
      assert !this.done;
      this.done = true;
      notifyAll();
    }

    private synchronized void noMoreRead() {
      noMoreRead = true;
      notifyAll();
    }

    private synchronized List<byte[]> getNextList() throws InterruptedException {
      while (!done && next == null) {
        wait();
      }
      List<byte[]> ret = next;
      next = null;

      notifyAll();
      return ret;
    }

    /**
     * .
     * NOTE: thread unsafe
     */
    private class ValueIterator implements CloseableIterator<byte[]> {
      private Iterator<byte[]> cur = null;

      @Override
      public boolean hasNext() {
        try {
          while (cur == null || !cur.hasNext()) {
            List<byte[]> n = getNextList();
            if (n != null) {
              cur = n.iterator();
            } else {
              cur = null;
              break;
            }
          }
          return cur != null;
        } catch(InterruptedException e) {
          throw new RuntimeException(e);
        }
      }

      @Override
      public byte[] next() {
        if (!hasNext()) {
          throw new NoSuchElementException("no more key/value pair in local.");
        }
        return cur.next();
      }

      @Override
      public void remove() {
        throw new UnsupportedOperationException();
      }

      @Override
      public void close() {
        noMoreRead();
      }
    };

    public CloseableIterator<byte[]> iterator() {
      return new ValueIterator();
    }
  }
}
